package ai.leinao.common.core.actor;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.Objects;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * swp actor 邮箱
 * 一个业务有一个Actor ,一个 Actor 有一个唯一 MailBox
 *
 * @author zhong.wang
 * @date 2025/4/3
 */
@Slf4j
@Data
public abstract class AbstractSwpActorMailbox implements SwpActorMailbox {

    /**
     * 高优先级
     */
    private static final boolean HIGH_PRIORITY = true;
    /**
     * 低优先级
     */
    private static final boolean NORMAL_PRIORITY = false;

    /**
     * 邮箱状态
     */
    private static final boolean FREE = false;
    private static final boolean BUSY = true;

    /**
     * 邮箱状态
     */
    private static final boolean NOT_READY = false;
    private static final boolean READY = true;

    /**
     * 忙碌状态
     * function: 控制并发处理
     * FREE(false)：空闲状态，可获取处理权
     * BUSY(true)：正在处理消息中
     */
    private final AtomicBoolean busy = new AtomicBoolean(FREE);

    /**
     * 就绪状态(提供 子类或其他方法控制邮箱激活/冻结状态)
     * function: 控制邮箱是否允许处理消息队列
     * NOT_READY(false): 暂定处理消息
     * READY(true): 运行处理消息
     */
    private final AtomicBoolean ready = new AtomicBoolean(NOT_READY);

    /**
     * 高优先级消息队列
     */
    private final ConcurrentLinkedQueue<SwpActorMessage> highPriorityMessageQueue = new ConcurrentLinkedQueue<>();

    /**
     * 普通优先级消息队列
     */
    private final ConcurrentLinkedQueue<SwpActorMessage> normalPriorityMessageQueue = new ConcurrentLinkedQueue<>();

    /**
     * 吞吐量
     */
    private final int actorThroughput;

    /**
     * 执行器
     */
    private final ExecutorService executor;

    public int getThroughput() {
        return actorThroughput;
    }

    /**
     * 接收消息
     * @param message
     */
    public void receive(SwpActorMessage message){
        enqueue(message, NORMAL_PRIORITY);
    }

    /**
     * 接收高优先级消息
     * @param message
     */
    public void receiveWithHighPriority(SwpActorMessage message){
        enqueue(message, HIGH_PRIORITY);
    }


    /**
     * 入队
     * @param msg
     * @param highPriority
     */
    private void enqueue(SwpActorMessage msg, boolean highPriority){
        if (highPriority){
            highPriorityMessageQueue.add(msg);
        }else {
            normalPriorityMessageQueue.add(msg);
        }
        tryProcessQueue(true);
    }

    /**
     * 尝试处理队列
     * @param newMessage
     */
    private void tryProcessQueue(boolean newMessage) {
        if (ready.get() == READY) {
            if (newMessage || !highPriorityMessageQueue.isEmpty() || !normalPriorityMessageQueue.isEmpty()) {
                // 如果邮箱空闲，则尝试获取处理权
                // 同一时间只有一个线程能进入消息处理流程
                if (busy.compareAndSet(FREE, BUSY)) {
                    executor.execute(() -> {
                        processMailbox();
                    });
                }
            }
        }

    }

    /**
     * 处理消息
     */
    private void processMailbox(){
        boolean noMoreElements = false;
        for (int i = 0; i < getActorThroughput(); i++){
            SwpActorMessage msg = highPriorityMessageQueue.poll();
            if (Objects.isNull(msg)){
                msg = normalPriorityMessageQueue.poll();
            }
            if(Objects.nonNull(msg)){
                try {
                    log.debug("going to process message: {}", msg);
                    processMessage(msg);
                } catch (Throwable t) {
                    log.debug("failed to process message: {}", msg, t);
                }
            } else {
                noMoreElements = true;
                break;
            }
        }
        if (noMoreElements) {
            busy.set(FREE);
            getExecutor().execute(() -> tryProcessQueue(false));
        } else {
            getExecutor().execute(this::processMailbox);
        }

    }


}
